package com.cqs.learning.spring.cloud.producer.bean.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cqs.learning.spring.cloud.producer.bean.dto.HousePriceChannel;
import com.cqs.learning.spring.cloud.producer.bean.po.HousePrice;
import com.cqs.learning.spring.cloud.producer.service.HousePriceService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author li
 * @Date 5/31/20 11:25 PM
 */
//@EnableBinding(value = HousePriceChannel.class)
@Slf4j
@Deprecated
public class HousePriceStreamConsumer1 implements InitializingBean {


    @Resource
    HousePriceService housePriceService;

    private final static int MAX_CAPACITY = 20;
    private final static BlockingQueue<Runnable> housePriceQueue = new ArrayBlockingQueue<Runnable>(MAX_CAPACITY);

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS,
            housePriceQueue, r -> new Thread(r, "house-price-"), new ThreadPoolExecutor.AbortPolicy());
    private Random rand = new Random();

    /**
     * @param message
     */
    @StreamListener(HousePriceChannel.INPUT)
    public void consumerHousePriceMsg(Message<String> message) {
        String payload = message.getPayload();
        log.info("订阅House Price消息：{},housePriceService:{}", payload, housePriceService);
        try {
            //错误用法，这个队列不是用来给业务代码提任务的
            housePriceQueue.put(() -> subTask(payload));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (housePriceQueue.size() > (MAX_CAPACITY * 0.75)) {
            log.warn("队列中任务数量过多,进入休眠状态");
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    public void subTask(String payload) {
        //1. 转化为对象
        HousePrice housePrice = JSON.parseObject(payload, HousePrice.class);
        JSONObject jo = JSON.parseObject(payload);
        housePrice.setTui(jo.getString("Transaction unique identifier"));
        long start = System.currentTimeMillis();
        if (housePrice.getTransferDate() == null) {
            log.warn("不合法的数据{}:", payload);
            return;
        }
        boolean save = housePriceService.save(housePrice);
        long cost = System.currentTimeMillis() - start;
        if (!save) {
            log.warn("保存失败:{},cost:{}", payload, cost);
        } else {
            if (rand.nextInt(100) < 10) {
                log.info("保存成功:id{},cost:{}", housePrice.getId(), cost);
            }
        }
    }

    /**
     * 消费消息
     *
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() {

        Runnable r = () -> {
            while (true) {
                try {
                    Runnable poll = housePriceQueue.take();
                    //会包一层FutureTask, 结合上面的逻辑的，这儿会从队列中取出，包一层FutureTask，然后再放回队列尾部，如此循环，知道有线程消费这个任务
                    //因为生成大量的FutureTask，而且相互包裹(引用)，导致出现频繁的FGC
                    executor.submit(poll);
                    if (rand.nextInt(100) < 6) {
                        long taskCount = executor.getTaskCount();
                        long queueSize = executor.getQueue().size();
                        long completedTaskCount = executor.getCompletedTaskCount();
                        log.info("队列的任务taskCount:{}:size:{},complete task:{}", taskCount,
                                queueSize, completedTaskCount);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        new Thread(r, "kafka-task-submit").start();
    }

//    public static void main(String[] args) {
//        new Thread(() -> {
//            while (true) {
//                try {
//                    Runnable take = housePriceQueue.take();
//                    executor.submit(take);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//            }
//        }).start();
//
//        try {
//            TimeUnit.SECONDS.sleep(1);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        int cnt0 = 100;
//        while (--cnt0 >= 0) {
//            try {
//                if (housePriceQueue.size() < MAX_CAPACITY * 0.5) {
//                    housePriceQueue.put(new Task0());
//                } else {
//                    Thread.sleep(2000);
//                }
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            log.info(cnt + "\t" + housePriceQueue.size());
//        }
//
//        while (true) {
//            log.warn("队列大小" + "\t" + housePriceQueue.size());
//            try {
//                Thread.sleep(1000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
//
//    }

    private static AtomicInteger cnt = new AtomicInteger();

    static class Task0 implements Runnable {

        private int id;

        @Override
        public void run() {
            int i = cnt.addAndGet(1);
            id = i;
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info(Thread.currentThread().getId() + "第{}个任务结束", i);
        }
    }

    public static List<Integer> partitionLabels(String S) {
        int[] rightIdx = new int[26];
        char[] chars = S.toCharArray();
        for (int i = 0; i < chars.length; ++i) {
            rightIdx[chars[i] - 'a'] = i;
        }
        List<Integer> list = new ArrayList<>();
        int left = 0, right = 0;
        for (int idx = 0; idx < chars.length; idx++) {
            right = Math.max(rightIdx[chars[idx] - 'a'], right);
            //生成一个片段
            if (idx == right) {
                list.add(right - left + 1);
                left = idx + 1;
            }
        }
        return list;
    }

    public static void main(String[] args) {
        List<Integer> partitionLabels = partitionLabels("ababcbacadefegdehijhklij");
        System.out.println(partitionLabels);
    }

}